package com.west.xu.tdmq.handle;

import com.google.gson.GsonBuilder;
import com.west.xu.tdmq.domain.Notice;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageListener;

import java.util.concurrent.atomic.AtomicInteger;

public class ConsumerMsg<T> implements MessageListener<T> {

    private static final Log logger = LogFactory.getLog(ConsumerMsg.class);

    private final AtomicInteger ai = new AtomicInteger(0);
    private final int maxRedeliveryCount = 3;

    @Override
    public void received(Consumer<T> consumer, Message<T> message) {
        try {
//            String jsonMessageId = new GsonBuilder().create().toJson(messageId.getMessageId());
            String messageId = MessageId.fromByteArray(message.getMessageId().toByteArray()).toString();
            String msg = new String(message.getData());
            Notice notice =new GsonBuilder().create().fromJson(msg,Notice.class);
            int redeliveryCount = message.getRedeliveryCount();
            logger.info("接收到消息===>" + msg + "重投次数[" + redeliveryCount + "]|MessageId===>" + messageId);
            if ("retry".equals(notice.getMessage())) {
                if (redeliveryCount == maxRedeliveryCount) {
                    consumer.acknowledge(message);
                    logger.info("接收消息处理成功，应答成功");
                } else {
                    consumer.negativeAcknowledge(message);
                    logger.info("接收消息拒绝应答，重投消息");
                }
            } else if ("dead".equals(notice.getMessage())) {
                consumer.negativeAcknowledge(message);
                logger.info("接收消息拒绝应答，重投消息");
            } else {
                consumer.acknowledge(message);
                logger.info("接收消息处理成功，应答成功");
            }
        } catch (Exception e) {
            consumer.negativeAcknowledge(message);
            logger.error("接收消息异常，拒绝应答|Ex==" + e.getMessage());
        }
    }
}
